package com.fanxuankai.boot.canal.listener.autoconfigure;

import com.fanxuankai.boot.canal.listener.MessageListener;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;

import java.util.Set;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({RabbitTemplate.class})
@ConditionalOnMissingClass({"com.fanxuankai.boot.mqbroker.producer.EventPublisher"})
@ConditionalOnProperty(name = CanalListenerProperties.ENABLED, matchIfMissing = true)
public class CanalListenerRabbitMqAutoConfiguration {
    @Bean
    public MessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory,
                                                             AmqpAdmin amqpAdmin,
                                                             MessageListener messageListener) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        Set<String> topics = messageListener.getTopics();
        // 监听的队列
        for (String topic : topics) {
            Queue queue = new Queue(topic);
            amqpAdmin.declareQueue(queue);
            container.addQueues(queue);
        }
        // 设置监听器
        container.setMessageListener(message -> {
            String msg = new String(message.getBody());
            MessageProperties messageProperties = message.getMessageProperties();
            messageListener.onMessage(messageProperties.getConsumerQueue(), msg);
        });
        return container;
    }
}
